package com.niit.service

import com.niit.bean.Answer
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.DStream

class EDUDataService {


  def dataAnlysis(answer:DStream[Answer]): Unit ={
      hotQuestionTop10(answer)
      gradeQuestionTop10(answer)
      hotQuestionWithSubjectTop10(answer)
      studentMinScoreWithQuestionTop10(answer)
      subjectWithQuestionTop10(answer)
      studentMinScoreTop10Question(answer)
  }

  /*
   需求一：统计 前十（top10）的热点题  统计一个采集周期（5s） 出现次数最多的前10道题
    {"student_id":"学生ID_39",
  "textbook_id":"教材ID_1",
  "grade_id":"年级ID_4",
  "subject_id":"科目ID_2_语文",
  "chapter_id":"章节ID_chapter_3",
  "question_id":"题目ID_660",
  "score":4,
  "answer_time":"2024-05-27 09:51:38",
  "ts":"May 27, 2024 9:51:38 AM"}
   */
  private def hotQuestionTop10(answer:DStream[Answer]): Unit ={
    //1. key:question_id           value:总次数
    //(古诗词第一题,1) (应用题第一题,1)  (古诗词第一题,1)
    val mapDS: DStream[(String, Int)] = answer.map(line => {
      val question_id = line.question_id
      (question_id, 1)
    })
    //(古诗词第一题,2) (应用题第一题,1)
    val reduceDS: DStream[(String, Int)] = mapDS.reduceByKey(_ + _)

    reduceDS.foreachRDD(rdd=>{//rdd：一组的 DStream 转换成  一组 rdd
      val top10: Array[(String, Int)] = rdd.sortBy(_._2, false).take(10)
      //如果后续要根据此数据进行离线分析，可以将次数保存进数据库。这里我仅作演示，保存进数据库自行编写
      println("------------------需求一：统计 前十（top10）的热点题  统计一个采集周期（5s） 出现次数最多的前10道题-------------------")
      top10.foreach( println )
    })
  }
  /*
  需求二：统计top3答题活跃的年级  答题次数最多的前三个年级
   {"student_id":"学生ID_39",
  "textbook_id":"教材ID_1",
  "grade_id":"年级ID_4",
  "subject_id":"科目ID_2_语文",
  "chapter_id":"章节ID_chapter_3",
  "question_id":"题目ID_660",
  "score":4,
  "answer_time":"2024-05-27 09:51:38",
  "ts":"May 27, 2024 9:51:38 AM"}
   */
  private def gradeQuestionTop10(answer:DStream[Answer]): Unit ={
    val mapDS: DStream[(String, Int)] = answer.map(line => {
      val grade_id = line.grade_id
      (grade_id, 1)
    })
    //(古诗词第一题,2) (应用题第一题,1)
    val reduceDS: DStream[(String, Int)] = mapDS.reduceByKey(_ + _)

    reduceDS.foreachRDD(rdd=>{//rdd：一组的 DStream 转换成  一组 rdd
      val top10: Array[(String, Int)] = rdd.sortBy(_._2, false).take(3)
      //如果后续要根据此数据进行离线分析，可以将次数保存进数据库。这里我仅作演示，保存进数据库自行编写
      println("------------------需求二：统计top3答题活跃的年级-------------------")
      top10.foreach( println )
    })
  }
    /*
  需求三：统计Top10热点题并带上所属科目
   {"student_id":"学生ID_39",
  "textbook_id":"教材ID_1",
  "grade_id":"年级ID_4",
  "subject_id":"科目ID_2_语文",
  "chapter_id":"章节ID_chapter_3",
  "question_id":"题目ID_660",
  "score":4,
  "answer_time":"2024-05-27 09:51:38",
  "ts":"May 27, 2024 9:51:38 AM"}
   */
  private def hotQuestionWithSubjectTop10(answer:DStream[Answer]): Unit ={
    //1. Key  （question_id subject_id）           value ：次数
    /*
              question_id                subject_id               value
            古诗词填空第一题：1               语文                    1
             应用题第一题：1                  数学                    1
             古诗词填空第一题：1               语文                   1
     */
    val reduceDS: DStream[((String, String), Int)] = answer.map(line => {

      ((line.question_id, line.subject_id), 1)

    }).reduceByKey(_ + _) // 获得 科目中每个题目出现总次数

    reduceDS.foreachRDD(rdd=>{
      val top10: Array[((String, String), Int)] = rdd.sortBy(_._2, false).take(10)
      println("------------------ 需求三：统计Top10热点题并带上所属科目-------------------")
      top10.foreach( println )
    })

  }

  /*
  需求四：统计Top10 学生最得分最低的成绩，并带上题目
   */
  private def studentMinScoreWithQuestionTop10(answer:DStream[Answer]): Unit ={
    //key student  question  value: score
    val mapDS: DStream[((String, String), Int)] = answer.map(line => {
      ((line.student_id, line.subject_id), line.score)
    })

    mapDS.foreachRDD(rdd=>{
      val top10: Array[((String, String), Int)] = rdd.sortBy(_._2, true).take(10)
      println("------------------ 需求四：统计Top10 学生最得分最低的成绩，并带上题目-------------------")
      top10.foreach( println )
    })

  }

  /*
    需求五：实时统计 每个科目的 Top10 热点 reduceByKey groupByKey

     语文： 古诗词第一题-100    古诗词第二题-89  古诗词第三题-74
   */
  private def subjectWithQuestionTop10(answer:DStream[Answer]): Unit ={
    // key subject_id   value:[(question_id_1,10),……………………((question_id_10,2))]
    val mapDS: DStream[((String, String), Int)] = answer.map(line => {
      ((line.subject_id, line.question_id), 1)
    }) // 科目下每道题出现的次数
    val reduceDS: DStream[((String, String), Int)] = mapDS.reduceByKey(_ + _)// 科目下每道题出现的总次数
    //[ ((语文,古诗词第一题),100),   ((语文,古诗词第二题),89) ,((语文,古诗词第三题),74)  ]

    val newMapDS: DStream[(String, (String, Int))] = reduceDS.map {
      case ((subjectId, questionId), count) => {
        (subjectId, (questionId, count))
      }
    }
    //[ (语文,(古诗词第一题,100)),   (语文,(古诗词第二题,89)) ,(语文,(古诗词第三题,74))  ]
    /*
      reduceByKey:对相同Key的Value 进行相加   ("北海",200) ("北海",300),("北海",400)  reduceByKey => ("北海",900)
      groupByKey：对相同Key 的Value,进行拼接  ("北海",200) ("北海",300),("北海",400)  groupByKey =>("北海",[200,300,400])
     */
    val groupDS: DStream[(String, Iterable[(String, Int)])] = newMapDS.groupByKey()
    groupDS.foreachRDD(rdd=>{
      /*
         map ：得到数据是整体数据 包括 key  value
         mapvalue:得到数据是部分数据  只包括 value
       */
      val top10: RDD[(String, List[(String, Int)])] = rdd.mapValues(value => {
        val top10: List[(String, Int)] = value.toList.sortBy(_._2).reverse.take(10)
        top10
      })
      println("------------------ 需求五：实时统计 每个科目的 Top10 热点题-------------------")
      top10.foreach( println )
    })

  }

    /*
   需求六：统计每个学生得分最低的前10道题，并带上所属题目
     */
  private def studentMinScoreTop10Question(answer:DStream[Answer]): Unit ={
    // Key:  studentId value :question_Id  score
    val mapDS: DStream[(String, (String, Int))] = answer.map(line => {
      (line.student_id, (line.question_id, line.score))
    })
    //[ (张三,(题目1,3)),   (张三,(题目2,4)),            (张三,(题目3,4)),..........        (张三,(题目99,1))    ]
    val groupDS: DStream[(String, Iterable[(String, Int)])] = mapDS.groupByKey()

    groupDS.foreachRDD(rdd=>{
      val top10: RDD[(String, List[(String, Int)])] = rdd.mapValues(value => {
        val top10: List[(String, Int)] = value.toList.sortBy(_._2).take(10)
        top10
      })
      println("------------------ 需求六：统计每个学生得分最低的前10道题，并带上所属题目-------------------")
      top10.foreach( println )
    })
  }
}
